package com.pisolution.scheduler.remote;

import com.pisolution.scheduler.remote.codec.CommandDecoder;
import com.pisolution.scheduler.remote.codec.CommandEncoder;
import com.pisolution.scheduler.remote.command.CommandType;
import com.pisolution.scheduler.remote.handler.NettyServerHandler;
import com.pisolution.scheduler.remote.processor.NettyRequestProcessor;
import com.pisolution.scheduler.remote.utils.NettyUtils;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 节点组件之间信息交互的服务端.
 *
 * @author sora
 * @version 0.1.0
 */
public final class NettyRemoteServer {
    private static final Logger LOGGER = LoggerFactory.getLogger(NettyRemoteServer.class);

    /** 用于创建服务端事件组. */
    private static final int CPUS = Runtime.getRuntime().availableProcessors();

    // -------------------------------------------------------------------------------------------------

    /** 用于配置并启动服务端. */
    private final ServerBootstrap bootstrap;

    /** 用于接受客户端连接的事件组. */
    private final EventLoopGroup acceptorGroup;

    /** 用于响应客户端请求的事件组. */
    private final EventLoopGroup clientGroup;

    /** 用于创建客户端请求处理的环境. */
    private final ExecutorService executor;

    /** 用于路由请求到适配的处理器. */
    private final NettyServerHandler handler;

    /** 用于编码交互命令. */
    private final CommandEncoder encoder;

    // -------------------------------------------------------------------------------------------------

    /** 服务端对外开放访问的端口. */
    private final int port;

    // -------------------------------------------------------------------------------------------------

    /** 用于确认服务端程序是否已经启动. */
    private final AtomicBoolean isStarted = new AtomicBoolean(false);

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化服务端配置.
     *
     * @param port 服务端程序启动时所使用的端口.
     */
    public NettyRemoteServer(final int port) {
        // 1. 初始化服务端配置.
        this.bootstrap = new ServerBootstrap();
        this.executor = Executors.newFixedThreadPool(CPUS);
        this.handler = new NettyServerHandler(this.executor);
        this.encoder = new CommandEncoder();

        this.port = port;

        // 2. 如果epoll技术可用 则启用该技术.
        if (NettyUtils.useEpoll()) {
            this.acceptorGroup = new EpollEventLoopGroup(1);
            this.clientGroup = new EpollEventLoopGroup(CPUS);

            return;
        }

        this.acceptorGroup = new NioEventLoopGroup(1);
        this.clientGroup = new NioEventLoopGroup(CPUS);
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    /**
     * 配置并启动服务端.
     */
    public void start() {
        if (this.isStarted.compareAndSet(false, true)) {
            // 1. 配置服务端.
            this.bootstrap
                .group(this.acceptorGroup, this.clientGroup)
                .channel(NettyUtils.useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, 65535)
                .childOption(ChannelOption.SO_RCVBUF, 65535)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) {
                        channel
                            .pipeline()
                            .addLast("encoder", encoder)
                            .addLast("decoder", new CommandDecoder())
                            .addLast("handler", handler);
                    }
                });

            // 2. 尝试启动服务端.
            ChannelFuture future;

            try {
                future = this.bootstrap.bind(this.port).sync();
            } catch (Exception e) {
                LOGGER.error("Could not bind the remote server to the port {} with cause {}.",
                    this.port,
                    e.getMessage(),
                    e
                );

                throw new RuntimeException(String.format("Could not bind the remote server to the port %d.", this.port));
            }

            // 3. 确认服务端是否已经启动.
            if (future.isSuccess()) {
                LOGGER.info("The remote server started at port {}.", this.port);

                return;
            }

            // 4. 确认启动失败原因.
            if (future.cause() != null) {
                throw new RuntimeException(
                    String.format("Could not bind the remote server to the port %d.", this.port),
                    future.cause()
                );
            }

            throw new RuntimeException(String.format("Could not bind the remote server to the port %d.", this.port));
        }
    }

    /**
     * 关闭服务端.
     */
    public void stop() {
        if (this.isStarted.compareAndSet(true, false)) {
            try {
                // 1. 关闭客户端连接接受事件组.
                if (this.acceptorGroup != null) {
                    this.acceptorGroup.shutdownGracefully();
                }

                // 2. 关闭客户端请求响应事件组.
                if (this.clientGroup != null) {
                    this.clientGroup.shutdownGracefully();
                }

                // 3. 关闭执行器服务.
                if (this.executor != null) {
                    this.executor.shutdown();
                }
            } catch (Exception e) {
                LOGGER.error("Could not close the remote server.", e);
            }

            LOGGER.info("The remote server closed.");
        }
    }

    // -------------------------------------------------------------------------------------------------
    //  处理器注册
    // -------------------------------------------------------------------------------------------------

    /**
     * 注册给定的处理器到服务端以处理相应的请求命令 处理时使用给定的执行器创建执行环境.
     *
     * @param type 该处理器所处理的请求命令类型.
     * @param processor 待注册处理器.
     *
     * @return {@link NettyRemoteServer} 对象以便于链式调用注册多个处理器.
     */
    public NettyRemoteServer register(final CommandType type, final NettyRequestProcessor processor) {
        return this.register(type, processor, null);
    }

    /**
     * 注册给定的处理器到服务端以处理相应的请求命令 处理时使用给定的执行器创建执行环境.
     *
     * @param type 该处理器所处理的请求命令类型.
     * @param processor 待注册处理器.
     * @param executor 用于创建处理器的执行环境.
     *
     * @return {@link NettyRemoteServer} 对象以便于链式调用注册多个处理器.
     */
    public NettyRemoteServer register(final CommandType type, final NettyRequestProcessor processor, final ExecutorService executor) {
        this.handler.register(type, processor, executor);

        return this;
    }
}
